package com.bubbles.workspace.sync.service.source.metadata;


import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.bubbles.engine.data.core.repository.BaseRepository;
import com.bubbles.engine.data.core.service.BaseService;
import com.bubbles.workspace.sync.constant.Constant;
import com.bubbles.workspace.sync.entity.source.metadata.SourceCatalog;
import com.bubbles.workspace.sync.entity.source.metadata.SourceJdbc;
import com.bubbles.workspace.sync.entity.source.metadata.SourceServer;
import com.bubbles.workspace.sync.key.source.metadata.SourceJdbcKey;
import com.bubbles.workspace.sync.key.source.metadata.SourceServerKey;
import com.bubbles.workspace.sync.properties.DorisProperties;
import com.bubbles.workspace.sync.properties.SshProperties;
import com.bubbles.workspace.sync.repository.source.metadata.SourceCatalogRepository;
import com.bubbles.workspace.sync.service.source.constraint.*;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDimensionService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeColumnService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeTableService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceIncrementService;
import com.bubbles.workspace.sync.service.source.feature.SourceTestService;
import com.bubbles.workspace.sync.util.SshManager;
import com.bubbles.workspace.sync.vo.SourceCatalogSaveVo;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * <p>description: 源数据源-目录 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@DS("pg")
@Slf4j
@Service
public class SourceCatalogService extends BaseService<SourceCatalog, String> {
    
    @Resource
    private SourceCatalogRepository sourceCatalogRepository;
    
    @Override
    public BaseRepository<SourceCatalog, String> getRepository() {
        return sourceCatalogRepository;
    }
    
    @Resource
    @Lazy
    private SourceTestService sourceTestService;
    
    @Resource
    private SourceJdbcService sourceJdbcService;
    
    @Resource
    private SourceServerService sourceServerService;
    
    @Resource
    private SourceCatalogService sourceCatalogService;
    
    @Resource
    private SourceDatabaseService sourceDatabaseService;
    
    @Resource
    private SourceTableService sourceTableService;
    
    @Resource
    private SourceColumnService sourceColumnService;
    
    @Resource
    private SourceConstraintPrimaryService sourceConstraintPrimaryService;
    
    @Resource
    private SourceConstraintUniqueService sourceConstraintUniqueService;
    
    @Resource
    private SourceConstraintForeignService sourceConstraintForeignService;
    
    @Resource
    private SourceConstraintCheckService sourceConstraintCheckService;
    
    @Resource
    private SourceIndexUniqueService sourceIndexUniqueService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeTableService sourceEnhanceDorisAnalyzeTableService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeColumnService sourceEnhanceDorisAnalyzeColumnService;
    
    @Resource
    private SourceEnhanceIncrementService sourceEnhanceIncrementService;
    
    @Resource
    private SourceEnhanceDimensionService sourceEnhanceDimensionService;
    
    @Resource
    private SshProperties sshProperties;
    
    @Resource
    private SshManager sshManager;
    
    @Resource
    private DorisProperties dorisProperties;
    
    @Transactional
    public List<SourceCatalog> listEnabled() {
        return sourceCatalogRepository.listEnabled();
    }
    
    @DSTransactional
    public void addVo(SourceCatalogSaveVo vo) {
        SourceServer sourceServer;
        SourceJdbc sourceJdbc;
        SourceCatalog sourceCatalog;
        
        // 目录名称唯一
        sourceCatalog = sourceCatalogService.getById(vo.getSourceCatalogName());
        if (sourceCatalog != null) {
            throw new RuntimeException("[Source] 源数据源目录已存在 => " + sourceCatalog.getSourceCatalogName());
        }
        
        // 系统编码唯一
        sourceCatalog = sourceCatalogRepository.getByTransformSystemCode(vo.getSourceSystemCode());
        if (sourceCatalog != null) {
            throw new RuntimeException("[Source] 转换系统编码冲突 => " + sourceCatalog.getSourceCatalogName());
        }
        
        // Server
        sourceServer = sourceServerService.getById(new SourceServerKey(vo.getHost(), vo.getPort()));
        if (sourceServer == null) {
            sourceServer = vo.createSourceServer();
            sourceServerService.save(sourceServer);
        }
        
        // JDBC
        sourceJdbc = sourceJdbcService.getById(new SourceJdbcKey(vo.getHost(), vo.getPort(), vo.getDatabaseType(), vo.getRoute(), vo.getUserName()));
        if (sourceJdbc == null) {
            sourceJdbc = vo.createSourceJdbc(sourceServer);
            sourceJdbcService.save(sourceJdbc);
        }
        
        // Catalog
        sourceCatalog = vo.createSourceCatalog(sourceJdbc);
        sourceCatalogService.save(sourceCatalog);
        
        // 测试
        sourceTestService.testCatalogConnection(sourceCatalog);
        
        // 同步数据库信息
        if (sourceCatalog.getSourceJdbc().getConnected()) {
            sourceDatabaseService.syncSource(sourceCatalog);
        }
    }
    
    @Transactional
    public void saveVo(SourceCatalogSaveVo vo) {
        SourceServer sourceServer;
        SourceJdbc sourceJdbc;
        SourceCatalog sourceCatalog;
        
        // 目录名称唯一
        sourceCatalog = sourceCatalogService.getById(vo.getSourceCatalogName());
        if (sourceCatalog == null) {
            throw new RuntimeException("[Source] 源数据源目录不存在 => " + vo.getSourceCatalogName());
        }
        
        // 系统编码唯一
        if (!StringUtils.equals(vo.getSourceSystemCode(), sourceCatalog.getTransform().getTransformSystemCode())) {
            sourceCatalog = sourceCatalogRepository.getByTransformSystemCode(vo.getSourceSystemCode());
            if (sourceCatalog != null && !StringUtils.equals(vo.getSourceCatalogName(), sourceCatalog.getSourceCatalogName())) {
                throw new RuntimeException("[Source] 转换系统编码冲突 => " + sourceCatalog.getSourceCatalogName());
            }
        }
        
        // 重新查询
        sourceCatalog = sourceCatalogService.getById(vo.getSourceCatalogName());
        SourceJdbc originSourceJdbc = sourceCatalog.getSourceJdbc();
        SourceServer originSourceServer = originSourceJdbc.getSourceServer();
        
        // Server
        sourceServer = sourceServerService.getById(new SourceServerKey(vo.getHost(), vo.getPort()));
        if (sourceServer == null) {
            sourceServer = vo.createSourceServer();
            sourceServerService.save(sourceServer);
        }
        
        // JDBC
        sourceJdbc = sourceJdbcService.getById(new SourceJdbcKey(vo.getHost(), vo.getPort(), vo.getDatabaseType(), vo.getRoute(), vo.getUserName()));
        if (sourceJdbc == null) {
            sourceJdbc = vo.createSourceJdbc(sourceServer);
        } else {
            if (StringUtils.isNotEmpty(vo.getPassword())) {
                sourceJdbc.setPassword(vo.getPassword());
            }
            if (!sourceJdbc.getConnectType().equals(vo.getConnectType())) {
                sourceJdbc.setConnectType(vo.getConnectType())
                          .setUrl(vo.getDatabaseType()
                                    .getDatabase()
                                    .getJdbcUrl(vo.getConnectType(), vo.getHost(), vo.getPort(), vo.getRoute()));
            }
            sourceJdbc = sourceJdbc.setSourceServer(sourceServer);
        }
        sourceJdbcService.save(sourceJdbc);
        
        // Catalog
        sourceCatalog = sourceCatalog.setSourceSystemCode(vo.getSourceSystemCode())
                                     .setServer(new SourceCatalog.Server().setHost(vo.getHost()).setPort(vo.getPort()))
                                     .setJdbc(new SourceCatalog.Jdbc().setDatabaseType(vo.getDatabaseType())
                                                                      .setRoute(vo.getRoute())
                                                                      .setUserName(vo.getUserName()))
                                     .setTransform(new SourceCatalog.Transform().setTransformSystemCode(StringUtils.lowerCase(vo.getSourceSystemCode())))
                                     .setSourceJdbc(sourceJdbc);
        sourceCatalogService.save(sourceCatalog);
        
        // 清空
        sourceJdbcService.clearWithoutRelation(originSourceJdbc);
        sourceServerService.clearWithoutRelation(originSourceServer);
        
        // 测试
        sourceTestService.testCatalogConnectionAsync(sourceCatalog);
    }
    
    @Transactional
    public String createCatalogSql(SourceCatalog sourceCatalog, boolean test) {
        String transformSystemCode = sourceCatalog.getTransform().getTransformSystemCode();
        if (StringUtils.isBlank(transformSystemCode)) {
            throw new RuntimeException("[Source] " + sourceCatalog.getSourceCatalogName() + " -> CREATE CATALOG -> 异常 => transformSystemCode=" + transformSystemCode);
        }
        
        SourceJdbc sourceJdbc = sourceCatalog.getSourceJdbc();
        String catalogName = (test ? Constant.TEST_CATALOG_PREFIX : "") + transformSystemCode;
        
        return String.format("""
                                     CREATE CATALOG IF NOT EXISTS `%s` COMMENT '%s'
                                     PROPERTIES (
                                         "type"="jdbc",
                                         "driver_url" = "%s",
                                         "driver_class" = "%s",
                                         "jdbc_url" = "%s",
                                         "user"="%s",
                                         "password"="%s",
                                         "lower_case_meta_names" = "false",
                                         "meta_names_mapping" = "",
                                         "only_specified_database" = "false",
                                         "include_database_list" = "",
                                         "exclude_database_list" = "",
                                         "connection_pool_min_size" = "%d",
                                         "connection_pool_max_size" = "%d",
                                         "connection_pool_max_wait_time" = "%d",
                                         "connection_pool_max_life_time" = "%d",
                                         "connection_pool_keep_alive" = "%b",
                                         "enable.auto.analyze" ="false"
                                     );"""
                , catalogName
                , sourceCatalog.getSourceCatalogName()
                , dorisProperties.getDriverJar(sourceJdbc.getDatabaseType())
                , dorisProperties.getDriverClass(sourceJdbc.getDatabaseType())
                , sourceJdbc.getUrl()
                , sourceJdbc.getUserName()
                , sourceJdbc.getPassword()
                , dorisProperties.getCatalog().getConnectionPool().getMinSize()
                , dorisProperties.getCatalog().getConnectionPool().getMaxSize()
                , dorisProperties.getCatalog().getConnectionPool().getMaxWaitTime()
                , dorisProperties.getCatalog().getConnectionPool().getMaxLifeTime()
                , dorisProperties.getCatalog().getConnectionPool().getKeepAlive()
        );
    }
    
    @Transactional
    public void enabledSourceCatalog(String sourceCatalogName, boolean enabled) {
        SourceCatalog sourceCatalog = sourceCatalogService.getById(sourceCatalogName);
        sourceCatalog.setSourceCatalogEnabled(enabled);
    }
}